/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.compression.VariableLengthUnsignedLongSequenceCodec;
import edu.ucsd.db.bassarisk.data.values.LongType;
import edu.ucsd.db.bassarisk.data.values.LongType.LongTools;
import edu.ucsd.db.bassarisk.data.values.LongType.ObjectLongReference;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;

public class VariableLengthLongColumnIndexReader implements
    IColumnIndexReader<LongType> {

  protected static class Cursor {
    protected int addressByteOffset;
    protected final ObjectLongReference drawerAddress = new ObjectLongReference();
    protected final ObjectLongReference drawerEntryDelta = new ObjectLongReference();
    protected int drawerIndex;
    protected long drawerStartEntry;
    protected int entryByteOffset;
  }

  private static final short columnIndexVersion = 4;
  private static final int HEADER_BYTES = 15;
  private ByteBuffer columnIndexBuffer;
  private IColumnSegmentInfo<LongType> columnInfo;
  private final Cursor current;
  private int dataDrawerCount;
  private final VariableLengthUnsignedLongSequenceCodec longCodec;

  public VariableLengthLongColumnIndexReader() {
    this.longCodec = VariableLengthUnsignedLongSequenceCodec.INSTANCE;
    LongTools.INSTANCE.getGenericCodec();

    this.current = new Cursor();
  }

  // For use in test
  public VariableLengthLongColumnIndexReader(final VariableLengthUnsignedLongSequenceCodec injectedCodec,
                                             final Cursor injectedCursor,
                                             final IColumnSegmentInfo<LongType> injectedDrawerInfo,
                                             final int injectedDataCardCount) {
    this.longCodec = injectedCodec;
    LongTools.INSTANCE.getGenericCodec();

    this.current = injectedCursor;
    this.columnInfo = injectedDrawerInfo;
    this.dataDrawerCount = injectedDataCardCount;
  }

  @Override
  public int getColumnDrawerCount() {
    return this.dataDrawerCount;
  }

  @Override
  public int getColumnIndexVersion() {
    return columnIndexVersion;
  }

  @Override
  public void initialize(final ByteBuffer columnIndexBuffer,
                         final ColumnReader<LongType>.ColumnInfo columnInfo,
                         final int headerCardAddress) {
    this.columnInfo = columnInfo;
    this.columnIndexBuffer = columnIndexBuffer;

    // Extract header information from the tail of the headerBuffer
    int headerPosition =
        ((headerCardAddress + 1) * ColumnStore.BYTES_IN_CARD) - HEADER_BYTES;
    this.columnIndexBuffer.position(headerPosition);

    int indexCardCount = this.columnIndexBuffer.get();
    this.dataDrawerCount = this.columnIndexBuffer.getShort();
    long entryCount = this.columnIndexBuffer.getLong();
    int bytesInEntryIndex = this.columnIndexBuffer.getShort();
    assert columnIndexVersion == this.columnIndexBuffer.getShort();
    this.columnIndexBuffer.clear();
    // Leave the headerBuffer in a cleared state

    int firstIndexCard = headerCardAddress - indexCardCount + 1;
    this.current.drawerIndex = -1;
    this.current.entryByteOffset = firstIndexCard * ColumnStore.BYTES_IN_CARD;
    this.current.addressByteOffset =
        this.current.entryByteOffset + bytesInEntryIndex;
    this.current.entryByteOffset +=
        this.longCodec.decode(this.columnIndexBuffer, this.current.drawerEntryDelta,
                              this.current.entryByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);

    assert this.current.drawerStartEntry == columnInfo.getFirstEntryIndex();
    
    columnInfo.reset(entryCount, 
                     new LongType.ObjectLongReference(Long.MAX_VALUE), 
                     new LongType.ObjectLongReference(Long.MIN_VALUE));
  }

  @Override
  public DrawerInfo<LongType> makeDrawerInfoInstance() {
    return new DrawerInfo<LongType>(LongTools.INSTANCE);
  }

  @Override
  public void setDrawerInfo(final int drawerIndex, final DrawerInfo<LongType> info) {
    assert drawerIndex >= this.current.drawerIndex;
    assert drawerIndex < this.dataDrawerCount;

    while (drawerIndex > this.current.drawerIndex) {
      this.stepCursor();
    }
    info.reset(this.current.drawerIndex, this.current.drawerAddress.getValue(),
               this.current.drawerEntryDelta.getValue(),
               this.current.drawerStartEntry);
  }

  @Override
  public boolean setDrawerInfoForEntryIndex(final int targetEntryIndex,
                                            final DrawerInfo<LongType> info) {
    assert targetEntryIndex >= this.current.drawerStartEntry; // forward only

    if (!this.columnInfo.hasEntry(targetEntryIndex)) { return false; }

    while (targetEntryIndex >= (this.current.drawerStartEntry + this.current.drawerEntryDelta
        .getValue())) {
      this.stepCursor();
    }
    info.reset(this.current.drawerIndex, this.current.drawerAddress.getValue(),
               this.current.drawerEntryDelta.getValue(),
               this.current.drawerStartEntry);
    return true;
  }

  @Override
  public boolean setDrawerInfoForValue(final IValueReference<LongType> value,
                                       final DrawerInfo<LongType> info) {
    String message = "Operation is only supported on sorted columns";
    throw new UnsupportedOperationException(message);
  }

  @Override
  public boolean setNextDrawerInfo(final DrawerInfo<LongType> info) {
    if ((this.current.drawerIndex + 1) == this.dataDrawerCount) { return false; }

    this.stepCursor();
    info.reset(this.current.drawerIndex, this.current.drawerAddress.getValue(),
               this.current.drawerEntryDelta.getValue(),
               this.current.drawerStartEntry);
    return true;
  }

  private void stepCursor() {
    this.current.drawerStartEntry += this.current.drawerEntryDelta.getValue();

    this.current.entryByteOffset +=
        this.longCodec.decode(this.columnIndexBuffer, this.current.drawerEntryDelta,
                              this.current.entryByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);
    this.current.addressByteOffset +=
        this.longCodec.decode(this.columnIndexBuffer, this.current.drawerAddress,
                              this.current.addressByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);
    this.current.drawerIndex++;
  }
}
